package com.demo.webflux.simple4;

import java.time.Duration;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class StatisticService {

	private static final UserVM EMPTY_USER = new UserVM();
	// 引入了userReposiroty、messageReposiroty，他们提供了和mongo集合的响应式通信
	private UserReposiroty userReposiroty;
	private MessageReposiroty messageReposiroty;

	// updateStatistic方法流式传输由UserStatistic视图模型对象表示的统计事件，同时需要一个由messageFlux参数表示的传入聊天信息流
	public Flux<UserStatistic> updateStatistic(Flux<ChatMessage> messageFlux) {
		return messageFlux
				.map(MessageMapper::toDomainUnit) // 订阅了ChatMessage对象的一个Flux，将他们转换为所需的表现形式，并使用messageReposiroty保存到mongodb
				.transform(messageReposiroty::saveAll)
				.retryBackoff(Long.MAX_VALUE, Duration.ofMillis(500)) // 有助于克服潜在的mongodb通信问题
				.onBackpressureLatest() // 此外，如果订阅者无法处理所有事件，我们会丢弃旧消息
				.concatMap(e -> this.doGetUserStatistic()); // 通过concatMap操作符，我们调用doGetUserStatistic方法来启动统计重计算过程。我们之所以使用concatMap，是因为它保证了统计结果的顺序正确，这是因为操作符会在生成下一个子流之前等待内部子流完成
	}

	// 计算顶级用户
	private Mono<UserStatistic> doGetUserStatistic() {
		Mono<UserVM> topActiveUserMono = userReposiroty
											.findMostActive() // 为了计算最活跃的用户，我们调用findMostActive方法，并将结果映射到正确的类型
											.map(UserMapper::toViewModelUnits)
											.defaultIfEmpty(EMPTY_USER); // 如果没有找到用户，就返回预定义的EMPTY_USER
		Mono<UserVM> topMentionedUserMono = userReposiroty
											.findMostPopular() // 为了获得最受欢迎的用户，我们调用findMostPopular方法
											.map(UserMapper::toViewModelUnits)
											.defaultIfEmpty(EMPTY_USER); // 如果没有找到用户，就返回预定义的EMPTY_USER
		return Mono
				.zip(topActiveUserMono, topMentionedUserMono, UserStatistic::new) // 合并这两个响应式请求并生成UserStatistic类的实例
				.timeout(Duration.ofSeconds(2)); // 设置可用于重新计算统计数据的最大时间预算
	}

}
